package main

import (
	"fmt"
	"github.com/golang/protobuf/proto"
	"github.com/withlin/canal-go/client"
	pbe "github.com/withlin/canal-go/protocol/entry"
	"gopkg.in/yaml.v3"
	"io/ioutil"
	"log"
	"os"
	"time"
)

type Config struct {
	Canal Canal `json:"canal" yaml:"canal"`
}

type Canal struct {
	Address     string `json:"address" yaml:"address"`
	Port        int    `json:"port" yaml:"port"`
	Username    string `json:"username" yaml:"username"`
	Password    string `json:"password" yaml:"password"`
	Destination string `json:"destination" yaml:"destination"`
	Timeout     int32  `json:"timeout" yaml:"timeout"`
	IdleTime    int32  `json:"idleTime" yaml:"idleTime"`
	ClientId    string `json:"clientId" yaml:"clientId"`
	Filter      string `json:"filter" yaml:"filter"`
}

func main() {
	// 读取 YAML 文件内容
	yamlFile, err := ioutil.ReadFile("../etc/config.yaml")
	if err != nil {
		fmt.Printf("Error reading YAML file: %v\n", err)
		return
	}

	// 定义一个 CanalConfig 结构体实例
	var config Config
	err = yaml.Unmarshal(yamlFile, &config)
	if err != nil {
		fmt.Printf("Error unmarshaling YAML data: %v\n", err)
		return
	}
	fmt.Printf("canal config: %v\n", config.Canal)

	/**
	  NewSimpleCanalConnector 参数说明
		client.NewSimpleCanalConnector("Canal服务端地址", "Canal服务端端口", "Canal服务端用户名", "Canal服务端密码", "Canal服务端destination", 60000, 60*60*1000)
	    Canal服务端地址：canal服务搭建地址IP
	    Canal服务端端口：canal\conf\canal.properties文件中
	    Canal服务端用户名、密码：canal\conf\example\instance.properties 文件中
	    Canal服务端destination ：canal\conf\example\meta.dat 文件中
	*/
	connector := client.NewSimpleCanalConnector(
		config.Canal.Address,
		config.Canal.Port,
		config.Canal.Username,
		config.Canal.Password,
		config.Canal.Destination,
		config.Canal.Timeout,
		config.Canal.IdleTime,
	)
	err = connector.Connect()
	if err != nil {
		log.Println(err)
		os.Exit(1)
	}

	// https://github.com/alibaba/canal/wiki/AdminGuide
	//mysql 数据解析关注的表，Perl正则表达式.
	//多个正则之间以逗号(,)分隔，转义符需要双斜杠(\\)
	//常见例子：
	//  1.  所有表：.*   or  .*\\..*
	//  2.  canal schema下所有表： canal\\..*
	//  3.  canal下的以canal打头的表：canal\\.canal.*
	//  4.  canal schema下的一张表：canal\\.test1
	//  5.  多个规则组合使用：canal\\..*,mysql.test1,mysql.test2 (逗号分隔)
	err = connector.Subscribe(config.Canal.Filter)
	if err != nil {
		log.Println(err)
		os.Exit(1)
	}

	for {

		message, err := connector.Get(100, nil, nil)
		if err != nil {
			log.Println(err)
			os.Exit(1)
		}
		batchId := message.Id
		if batchId == -1 || len(message.Entries) <= 0 {
			time.Sleep(300 * time.Millisecond)
			fmt.Println(time.Now().Format("2006-01-02 15:04:05"), "===没有数据了===")
			continue
		}

		printEntry(message.Entries)

	}
}

func printEntry(entrys []pbe.Entry) {

	for _, entry := range entrys {
		if entry.GetEntryType() == pbe.EntryType_TRANSACTIONBEGIN || entry.GetEntryType() == pbe.EntryType_TRANSACTIONEND {
			continue
		}
		rowChange := new(pbe.RowChange)

		err := proto.Unmarshal(entry.GetStoreValue(), rowChange)
		checkError(err)
		if rowChange != nil {
			eventType := rowChange.GetEventType()
			header := entry.GetHeader()
			fmt.Println(fmt.Sprintf("================> binlog[%s : %d],name[%s,%s], eventType: %s", header.GetLogfileName(), header.GetLogfileOffset(), header.GetSchemaName(), header.GetTableName(), header.GetEventType()))

			for _, rowData := range rowChange.GetRowDatas() {
				if eventType == pbe.EventType_DELETE {
					printColumn(rowData.GetBeforeColumns())
				} else if eventType == pbe.EventType_INSERT {
					printColumn(rowData.GetAfterColumns())
				} else {
					fmt.Println("-------> before")
					printColumn(rowData.GetBeforeColumns())
					fmt.Println("-------> after")
					printColumn(rowData.GetAfterColumns())
				}
			}
		}
	}
}

func printColumn(columns []*pbe.Column) {
	for _, col := range columns {
		fmt.Println(fmt.Sprintf("%s : %s  update= %t", col.GetName(), col.GetValue(), col.GetUpdated()))
	}
}

func checkError(err error) {
	if err != nil {
		fmt.Fprintf(os.Stderr, "Fatal error: %s", err.Error())
		os.Exit(1)
	}
}
